package jt.com.jt809.kafka;

import com.alibaba.fastjson2.JSON;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

import javax.annotation.Resource;

/**
 * kafka 操作类
 *
 * @Author OuJiaJun
 * @Date 2024/4/25 9:22
 */
@Slf4j
@Component
public class KafkaService {

    @Resource
    private KafkaTemplate<String, String> kafkaTemplate;

    public void sendMessageSync(String topic, BaseMode baseMode) {
        String jsonString = JSON.toJSONString(baseMode);
        ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send(topic, jsonString);
        future.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
            @Override
            public void onSuccess(SendResult<String, String> result) {
                log.info("发送成功：{}，Topic: {}, Partition: {}, Offset: {}",
                        jsonString, topic, result.getRecordMetadata().partition(), result.getRecordMetadata().offset());
            }
            @Override
            public void onFailure(Throwable ex) {
                log.error("发送异常：{}，Topic: {}", ex.getMessage(), topic);
                // 由于Kafka生产者已配置了重试次数，此处无需额外处理，仅记录异常信息即可
            }
        });
    }
}
